﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

namespace LS.Screening.UI.Common
{
    public static class RabbitMQConsumer
    {
        #region 01 订阅发布(广播消费) 扇形交换机

        /// <summary>
        /// 接收消息
        /// </summary>
        /// <param name="connection"></param>
        /// <param name="exchange"></param>
        /// <param name="action"></param>
        public static void ReceiveFanout(IConnection connection, string exchange, Action<string> action)
        {
            var channel = connection.CreateModel();
            channel.ExchangeDeclare(exchange: exchange, type: "fanout");

            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName, exchange: exchange, routingKey: "");
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body.ToArray());
                action?.Invoke(message);
                channel.BasicAck(ea.DeliveryTag, true);
            };
            channel.BasicQos(0, 1, false);
            channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
        }

        #endregion

        #region 02 主题交换机      

        public static void ReceiveTopic(string exchange, string queueName, string routingKey, Func<string, bool?> func)
        {
            var connection = PubSubMQConfig.GetConnection();
            var channel = connection.CreateModel();
            // 1、定义交换机
            channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Topic, true);
            channel.QueueDeclare(queueName, true, false, false, null);
            channel.QueueBind(queueName, exchange, routingKey);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body.ToArray());
                var flag = func?.Invoke(message);
                if (flag == true)
                    channel.BasicAck(ea.DeliveryTag, true);
            };
            // 3、消费消息
            channel.BasicQos(0, 1, false);
            channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
        }

        #endregion
    }
}
